Day 21: 週總結：專業領域 AI 助理

經過第三週的深度學習，我們掌握了自動化工作流程、條件判斷、並行處理、人機協作等進階技術。今天讓我們整合所有技能，打造一個專業領域的綜合 AI 助理系統！

📚 本週學習回顧

Day 15-20 技術棧掌握

🔄 自動化與流程

• ✅ 自動化工作流程設計（任務依賴、執行追蹤）
• ✅ 條件判斷與分支邏輯（智能路由、AI 決策）
• ✅ 並行處理與效能優化（多執行緒、快取機制）

🤝 互動與協作

• ✅ 人機協作介面設計（引導式工作流程、迭代優化）
• ✅ 視覺化回饋與進度追蹤（UI 組件、確認機制）

🎯 專業應用

• ✅ 資料分析助手（統計分析、AI 洞察、報告生成）
• ✅ 程式碼生成助手（智能生成、分析優化、測試生成）

🎯 專業領域 AI 助理設計

我們將建立一個「智能商務顧問助理」，整合本週所學的所有技術：

核心能力

• 📊 資料分析：分析業務數據，提供洞察
• 📝 文件生成：自動產生商業文件和報告
• 🔄 流程自動化：處理重複性商務任務
• 💬 智能對話：理解商業需求並提供建議
• 📈 趨勢預測：基於歷史數據預測未來趨勢

🏗 綜合系統架構

professional_ai_assistant/
├── main.py                              # 主程式入口
├── core/
│   ├── __init__.py
│   ├── assistant_engine.py              # 助理引擎核心
│   ├── workflow_orchestrator.py         # 工作流程協調器
│   └── knowledge_base.py                # 知識庫管理
├── modules/
│   ├── __init__.py
│   ├── business_analyzer.py             # 商業分析模組
│   ├── document_generator.py            # 文件生成模組
│   ├── task_automator.py                # 任務自動化模組
│   └── consultant_chat.py               # 顧問對話模組
├── integrations/
│   ├── __init__.py
│   ├── data_analysis_integration.py     # 資料分析整合
│   ├── code_generation_integration.py   # 程式碼生成整合
│   └── workflow_integration.py          # 工作流程整合
├── ui/
│   ├── __init__.py
│   ├── dashboard.py                     # 儀表板介面
│   └── interaction_components.py        # 互動組件
└── utils/
    ├── __init__.py
    ├── performance_tracker.py           # 效能追蹤器
    └── report_templates.py              # 報告模板

🔧 核心系統實作

1. 助理引擎核心 (core/assistant_engine.py)

import google.generativeai as genai
import os
from typing import Dict, Any, List
from datetime import datetime
import json

genai.configure(api_key=os.getenv('GEMINI_API_KEY'))

class ProfessionalAssistant:
    """專業領域 AI 助理引擎"""
    
    def __init__(self, domain: str = "business"):
        self.model = genai.GenerativeModel('gemini-2.5-flash')
        self.domain = domain
        self.conversation_history = []
        self.context = {
            'domain': domain,
            'session_start': datetime.now(),
            'user_profile': {},
            'active_tasks': []
        }
        
        # 載入領域知識
        self._load_domain_knowledge()
    
    def _load_domain_knowledge(self):
        """載入領域專業知識"""
        domain_knowledge = {
            'business': {
                'expertise': [
                    '商業策略規劃',
                    '市場分析',
                    '財務分析',
                    '營運優化',
                    '風險管理'
                ],
                'tools': [
                    'SWOT 分析',
                    'PESTEL 分析',
                    '五力分析',
                    '財務比率分析'
                ]
            },
            'technical': {
                'expertise': [
                    '系統架構設計',
                    '程式碼審查',
                    '效能優化',
                    '技術選型'
                ],
                'tools': [
                    '程式碼生成',
                    '架構圖生成',
                    '技術文件撰寫'
                ]
            }
        }
        
        self.knowledge = domain_knowledge.get(self.domain, {})
    
    def analyze_request(self, user_input: str) -> Dict[str, Any]:
        """分析使用者請求"""
        prompt = f"""
        作為專業的{self.domain}顧問，分析以下請求：
        
        使用者請求：{user_input}
        
        領域專長：{', '.join(self.knowledge.get('expertise', []))}
        可用工具：{', '.join(self.knowledge.get('tools', []))}
        
        請以 JSON 格式回應：
        {{
            "request_type": "analysis | generation | automation | consultation",
            "complexity": "simple | moderate | complex",
            "required_modules": ["模組名稱"],
            "estimated_time": "預估時間",
            "suggested_approach": "建議處理方式",
            "key_considerations": ["關鍵考量點"]
        }}
        
        只回傳 JSON。
        """
        
        try:
            response = self.model.generate_content(prompt)
            analysis = json.loads(response.text)
            
            print(f"🎯 請求類型：{analysis['request_type']}")
            print(f"📊 複雜度：{analysis['complexity']}")
            print(f"⏱️ 預估時間：{analysis['estimated_time']}")
            
            return analysis
        except Exception as e:
            print(f"⚠️ 分析失敗：{e}")
            return {
                "request_type": "consultation",
                "complexity": "moderate",
                "required_modules": [],
                "estimated_time": "未知"
            }
    
    def generate_professional_advice(self, topic: str, context: Dict = None) -> str:
        """生成專業建議"""
        context_str = ""
        if context:
            context_str = f"\n背景資訊：{json.dumps(context, ensure_ascii=False)}"
        
        prompt = f"""
        作為專業的{self.domain}顧問，針對以下主題提供深入的專業建議：
        
        主題：{topic}
        {context_str}
        
        請提供：
        1. 現況分析
        2. 關鍵問題識別
        3. 具體建議方案（至少 3 個）
        4. 實施步驟
        5. 預期效果
        6. 風險提示
        
        使用專業但易懂的語言，提供可執行的建議。
        """
        
        try:
            response = self.model.generate_content(prompt)
            return response.text
        except Exception as e:
            return f"建議生成失敗：{str(e)}"
    
    def coordinate_workflow(self, task_description: str) -> Dict[str, Any]:
        """協調多模組工作流程"""
        prompt = f"""
        請為以下任務設計工作流程：
        
        任務描述：{task_description}
        
        可用模組：
        - business_analyzer: 商業分析
        - document_generator: 文件生成
        - task_automator: 任務自動化
        - consultant_chat: 顧問諮詢
        
        請以 JSON 格式回應：
        {{
            "workflow_steps": [
                {{
                    "step": 1,
                    "module": "模組名稱",
                    "action": "動作描述",
                    "inputs": {{}},
                    "outputs": ["輸出1", "輸出2"]
                }}
            ],
            "parallel_tasks": ["可並行的任務"],
            "critical_path": ["關鍵路徑步驟"],
            "estimated_duration": "總時間"
        }}
        """
        
        try:
            response = self.model.generate_content(prompt)
            workflow = json.loads(response.text)
            return workflow
        except Exception as e:
            return {
                "error": str(e),
                "workflow_steps": []
            }

2. 商業分析模組 (modules/business_analyzer.py)

import pandas as pd
import numpy as np
from typing import Dict, Any, List
import google.generativeai as genai
import os
import json

genai.configure(api_key=os.getenv('GEMINI_API_KEY'))

class BusinessAnalyzer:
    """商業分析模組"""
    
    def __init__(self):
        self.model = genai.GenerativeModel('gemini-2.5-flash')
    
    def swot_analysis(self, business_info: Dict[str, Any]) -> Dict[str, List[str]]:
        """SWOT 分析"""
        prompt = f"""
        請針對以下企業進行 SWOT 分析：
        
        企業資訊：
        {json.dumps(business_info, ensure_ascii=False, indent=2)}
        
        請以 JSON 格式提供：
        {{
            "strengths": ["優勢1", "優勢2", "優勢3"],
            "weaknesses": ["劣勢1", "劣勢2", "劣勢3"],
            "opportunities": ["機會1", "機會2", "機會3"],
            "threats": ["威脅1", "威脅2", "威脅3"],
            "strategic_recommendations": ["建議1", "建議2"]
        }}
        """
        
        try:
            response = self.model.generate_content(prompt)
            return json.loads(response.text)
        except Exception as e:
            return {"error": str(e)}
    
    def market_analysis(self, market_data: Dict[str, Any]) -> str:
        """市場分析"""
        prompt = f"""
        請進行市場分析：
        
        市場資料：
        {json.dumps(market_data, ensure_ascii=False, indent=2)}
        
        分析重點：
        1. 市場規模和成長趨勢
        2. 目標客群特徵
        3. 競爭態勢
        4. 市場機會
        5. 進入策略建議
        
        提供專業的市場分析報告。
        """
        
        try:
            response = self.model.generate_content(prompt)
            return response.text
        except Exception as e:
            return f"分析失敗：{str(e)}"
    
    def financial_analysis(self, financial_data: pd.DataFrame) -> Dict[str, Any]:
        """財務分析"""
        # 計算關鍵財務指標
        analysis = {
            'revenue_growth': self._calculate_growth_rate(financial_data, 'revenue'),
            'profit_margin': self._calculate_profit_margin(financial_data),
            'liquidity_ratio': self._calculate_liquidity_ratio(financial_data),
            'key_insights': []
        }
        
        # 使用 AI 生成洞察
        insights_prompt = f"""
        基於以下財務指標，提供專業分析：
        
        營收成長率：{analysis['revenue_growth']:.2%}
        利潤率：{analysis['profit_margin']:.2%}
        流動比率：{analysis['liquidity_ratio']:.2f}
        
        請提供：
        1. 財務健康狀況評估
        2. 主要風險點
        3. 改善建議
        """
        
        try:
            response = self.model.generate_content(insights_prompt)
            analysis['ai_insights'] = response.text
        except:
            analysis['ai_insights'] = "洞察生成失敗"
        
        return analysis
    
    def _calculate_growth_rate(self, df: pd.DataFrame, column: str) -> float:
        """計算成長率"""
        if column not in df.columns or len(df) < 2:
            return 0.0
        
        first_value = df[column].iloc[0]
        last_value = df[column].iloc[-1]
        
        if first_value == 0:
            return 0.0
        
        return (last_value - first_value) / first_value
    
    def _calculate_profit_margin(self, df: pd.DataFrame) -> float:
        """計算利潤率"""
        if 'profit' in df.columns and 'revenue' in df.columns:
            total_profit = df['profit'].sum()
            total_revenue = df['revenue'].sum()
            return total_profit / total_revenue if total_revenue > 0 else 0.0
        return 0.0
    
    def _calculate_liquidity_ratio(self, df: pd.DataFrame) -> float:
        """計算流動比率"""
        if 'current_assets' in df.columns and 'current_liabilities' in df.columns:
            assets = df['current_assets'].iloc[-1]
            liabilities = df['current_liabilities'].iloc[-1]
            return assets / liabilities if liabilities > 0 else 0.0
        return 1.0

3. 工作流程協調器 (core/workflow_orchestrator.py)

from typing import Dict, Any, List, Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

class WorkflowOrchestrator:
    """工作流程協調器 - 整合本週所學技術"""
    
    def __init__(self):
        self.modules = {}
        self.execution_log = []
        self.cache = {}
    
    def register_module(self, name: str, module: Any):
        """註冊模組"""
        self.modules[name] = module
        print(f"✅ 已註冊模組：{name}")
    
    def execute_workflow(self, workflow: Dict[str, Any]) -> Dict[str, Any]:
        """執行工作流程（支援並行和條件判斷）"""
        steps = workflow.get('workflow_steps', [])
        parallel_tasks = workflow.get('parallel_tasks', [])
        
        print(f"\n🚀 開始執行工作流程（共 {len(steps)} 個步驟）")
        
        results = {}
        start_time = time.time()
        
        # 識別可並行的步驟
        parallel_step_ids = self._identify_parallel_steps(steps, parallel_tasks)
        
        # 按步驟順序執行
        for step in steps:
            step_id = step.get('step')
            
            # 檢查是否可並行執行
            if step_id in parallel_step_ids:
                # 收集所有並行步驟
                parallel_steps = [s for s in steps if s.get('step') in parallel_step_ids]
                
                if parallel_steps:
                    print(f"\n⚡ 並行執行 {len(parallel_steps)} 個步驟...")
                    parallel_results = self._execute_parallel(parallel_steps)
                    results.update(parallel_results)
                    
                    # 移除已執行的並行步驟
                    for ps in parallel_steps:
                        parallel_step_ids.discard(ps.get('step'))
            else:
                # 順序執行
                step_result = self._execute_single_step(step, results)
                results[f"step_{step_id}"] = step_result
        
        execution_time = time.time() - start_time
        
        return {
            'success': True,
            'results': results,
            'execution_time': execution_time,
            'steps_executed': len(steps)
        }
    
    def _identify_parallel_steps(self, steps: List[Dict], 
                                parallel_tasks: List[str]) -> set:
        """識別可並行執行的步驟"""
        parallel_ids = set()
        
        for step in steps:
            action = step.get('action', '')
            if any(task in action for task in parallel_tasks):
                parallel_ids.add(step.get('step'))
        
        return parallel_ids
    
    def _execute_parallel(self, steps: List[Dict]) -> Dict[str, Any]:
        """並行執行多個步驟"""
        results = {}
        
        with ThreadPoolExecutor(max_workers=3) as executor:
            future_to_step = {
                executor.submit(self._execute_single_step, step, {}): step
                for step in steps
            }
            
            for future in as_completed(future_to_step):
                step = future_to_step[future]
                try:
                    result = future.result()
                    results[f"step_{step['step']}"] = result
                    print(f"  ✅ 步驟 {step['step']} 完成")
                except Exception as e:
                    print(f"  ❌ 步驟 {step['step']} 失敗：{e}")
        
        return results
    
    def _execute_single_step(self, step: Dict, context: Dict) -> Any:
        """執行單個步驟"""
        module_name = step.get('module')
        action = step.get('action')
        inputs = step.get('inputs', {})
        
        print(f"📍 執行步驟 {step['step']}: {action}")
        
        # 檢查快取
        cache_key = f"{module_name}_{action}_{str(inputs)}"
        if cache_key in self.cache:
            print(f"  💾 使用快取結果")
            return self.cache[cache_key]
        
        # 執行模組方法
        if module_name in self.modules:
            module = self.modules[module_name]
            
            # 嘗試呼叫對應的方法
            if hasattr(module, action):
                method = getattr(module, action)
                result = method(**inputs)
                
                # 儲存到快取
                self.cache[cache_key] = result
                
                return result
        
        return {"status": "skipped", "reason": f"模組 {module_name} 不存在"}

4. 整合主程式 (main.py)

from core.assistant_engine import ProfessionalAssistant
from core.workflow_orchestrator import WorkflowOrchestrator
from modules.business_analyzer import BusinessAnalyzer
from ui.interaction_components import InteractionUI
import pandas as pd

def print_banner():
    """列印橫幅"""
    print("""
╔══════════════════════════════════════════════════════════════╗
║                                                              ║
║           🎯 專業領域 AI 助理系統                            ║
║                                                              ║
║     整合自動化、並行處理、人機協作、專業分析                 ║
║                                                              ║
╚══════════════════════════════════════════════════════════════╝
    """)

def main():
    """主程式"""
    print_banner()
    
    # 初始化系統
    assistant = ProfessionalAssistant(domain="business")
    orchestrator = WorkflowOrchestrator()
    business_analyzer = BusinessAnalyzer()
    ui = InteractionUI()
    
    # 註冊模組
    orchestrator.register_module("business_analyzer", business_analyzer)
    
    print("\n✨ 系統就緒！")
    print("\n📋 可用功能：")
    print("  1. 💼 商業分析（SWOT、市場分析、財務分析）")
    print("  2. 📝 專業建議（策略規劃、營運優化）")
    print("  3. 🔄 自動化工作流程")
    print("  4. 💬 智能顧問對話")
    
    while True:
        ui.print_separator()
        
        print("\n選擇服務：")
        print("1. 商業 SWOT 分析")
        print("2. 獲取專業建議")
        print("3. 執行自動化工作流程")
        print("4. 智能對話諮詢")
        print("5. 查看系統統計")
        print("6. 退出")
        
        try:
            choice = input("\n請選擇 (1-6)：").strip()
            
            if choice == '1':
                ui.print_header("商業 SWOT 分析")
                
                # 收集企業資訊
                company_info = ui.collect_business_info()
                
                if company_info:
                    print("\n🔍 分析中...")
                    swot = business_analyzer.swot_analysis(company_info)
                    
                    if 'error' not in swot:
                        ui.display_swot_analysis(swot)
                        
                        # 詢問是否產生報告
                        if ui.confirm("生成詳細分析報告"):
                            advice = assistant.generate_professional_advice(
                                f"{company_info['name']} 的 SWOT 分析",
                                swot
                            )
                            print(f"\n📄 分析報告：\n{advice}")
            
            elif choice == '2':
                ui.print_header("專業建議諮詢")
                
                topic = input("請描述您需要建議的主題：").strip()
                
                if topic:
                    print("\n💭 思考中...")
                    advice = assistant.generate_professional_advice(topic)
                    
                    ui.display_advice(advice)
            
            elif choice == '3':
                ui.print_header("自動化工作流程")
                
                task = input("請描述要自動化的任務：").strip()
                
                if task:
                    print("\n🎯 設計工作流程...")
                    
                    # 分析請求
                    analysis = assistant.analyze_request(task)
                    
                    # 協調工作流程
                    workflow = assistant.coordinate_workflow(task)
                    
                    if workflow.get('workflow_steps'):
                        print(f"\n📋 工作流程包含 {len(workflow['workflow_steps'])} 個步驟")
                        
                        if ui.confirm("執行此工作流程"):
                            result = orchestrator.execute_workflow(workflow)
                            
                            print(f"\n✅ 工作流程完成！")
                            print(f"⏱️ 執行時間：{result['execution_time']:.2f} 秒")
                            print(f"📊 執行步驟：{result['steps_executed']}")
            
            elif choice == '4':
                ui.print_header("智能顧問對話")
                
                print("💬 進入對話模式（輸入 'exit' 退出）")
                
                while True:
                    question = input("\n您：").strip()
                    
                    if question.lower() == 'exit':
                        break
                    
                    if question:
                        # 分析請求
                        analysis = assistant.analyze_request(question)
                        
                        # 生成回應
                        response = assistant.generate_professional_advice(
                            question,
                            {"complexity": analysis['complexity']}
                        )
                        
                        print(f"\n顧問：{response}")
            
            elif choice == '5':
                ui.print_header("系統統計")
                
                print(f"\n📊 系統運行統計：")
                print(f"  • 領域：{assistant.domain}")
                print(f"  • 會話時長：{(datetime.now() - assistant.context['session_start']).seconds // 60} 分鐘")
                print(f"  • 對話輪數：{len(assistant.conversation_history)}")
                print(f"  • 快取命中：{len(orchestrator.cache)} 項")
                print(f"  • 執行記錄：{len(orchestrator.execution_log)} 筆")
            
            elif choice == '6':
                if ui.confirm("確定要退出"):
                    print("\n👋 感謝使用專業領域 AI 助理！")
                    print("💡 期待下次為您服務")
                    break
            
            else:
                print("❌ 無效的選擇")
        
        except KeyboardInterrupt:
            print("\n\n👋 再見！")
            break
        except Exception as e:
            print(f"\n❌ 發生錯誤：{e}")
            print("請重試或選擇其他功能")

if __name__ == "__main__":
    from datetime import datetime
    main()

5. 互動UI組件 (ui/interaction_components.py)

from typing import Dict, Any, List

class InteractionUI:
    """互動UI組件"""
    
    def print_header(self, title: str):
        """列印標題"""
        print("\n" + "═" * 60)
        print(f"  {title}")
        print("═" * 60)
    
    def print_separator(self):
        """列印分隔線"""
        print("\n" + "─" * 60)
    
    def confirm(self, message: str) -> bool:
        """確認對話框"""
        response = input(f"\n❓ {message}？(y/n)：").strip().lower()
        return response in ['y', 'yes', '是']
    
    def collect_business_info(self) -> Dict[str, Any]:
        """收集企業資訊"""
        print("\n請提供企業基本資訊：")
        
        info = {}
        info['name'] = input("  企業名稱：").strip()
        info['industry'] = input("  產業類別：").strip()
        info['size'] = input("  企業規模 (小/中/大)：").strip() or "中"
        info['years'] = input("  成立年數：").strip() or "未知"
        info['description'] = input("  簡要描述：").strip()
        
        return info if info['name'] else None
    
    def display_swot_analysis(self, swot: Dict[str, List[str]]):
        """顯示 SWOT 分析結果"""
        self.print_header("SWOT 分析結果")
        
        categories = [
            ("💪 優勢 (Strengths)", swot.get('strengths', [])),
            ("⚠️  劣勢 (Weaknesses)", swot.get('weaknesses', [])),
            ("🌟 機會 (Opportunities)", swot.get('opportunities', [])),
            ("⚡ 威脅 (Threats)", swot.get('threats', []))
        ]
        
        for title, items in categories:
            print(f"\n{title}：")
            for i, item in enumerate(items, 1):
                print(f"  {i}. {item}")
        
        if 'strategic_recommendations' in swot:
            print(f"\n📋 策略建議：")
            for i, rec in enumerate(swot['strategic_recommendations'], 1):
                print(f"  {i}. {rec}")
    
    def display_advice(self, advice: str):
        """顯示建議內容"""
        print("\n┌" + "─" * 58 + "┐")
        
        lines = advice.split('\n')
        for line in lines:
            # 簡單的文字包裝
            if len(line) <= 56:
                print(f"│ {line:<56} │")
            else:
                words = line.split()
                current_line = ""
                for word in words:
                    if len(current_line) + len(word) + 1 <= 56:
                        current_line += word + " "
                    else:
                        print(f"│ {current_line:<56} │")
                        current_line = word + " "
                if current_line:
                    print(f"│ {current_line:<56} │")
        
        print("└" + "─" * 58 + "┘")

🎯 本週成就總結

技術整合度評估